package org.example.flinketl.util;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;

import javax.validation.ConstraintViolation;
import javax.validation.Validation;
import javax.validation.Validator;
import org.apache.flink.util.Collector;
import org.example.flinketl.vo.Vo;
import org.hibernate.validator.HibernateValidator;
import com.alibaba.fastjson.JSON;

public class MapAndFilter implements Serializable {
  static public int jsonFailCount = 0;
  static Validator validator = Validation.byProvider(HibernateValidator.class).configure().failFast(true)
      .buildValidatorFactory().getValidator();


  public static void flatmap(String value, Collector<Object> out) {
    // 根据类型转换vo
    JSON eventBody;
    String eventType;
    try{
      Map json =  JSON.parseObject(value, HashMap.class);
      eventBody = (JSON)json.get("eventBody");
      eventType = (String)json.get("eventType");
    }
    catch(Exception e){
      jsonFailCount++;
      return;
    }
    
    Object vo;
    try{
      Class<?> classType = MyConfig.tableTye.get(eventType);
      vo = JSON.parseObject(eventBody.toJSONString(), classType);
    }
    catch(Exception e){
      jsonFailCount++;
      return;
    }

    Set<ConstraintViolation<Object>> validate = validator.validate(vo);
    int result = validate.size();
    if (result != 0) {
      return;
    } else {
      out.collect(vo); 
    }

    // HashMap<String, Object> eveHashMap = JSON.parseObject(eventBody.toJSONString(), HashMap.class);
    // Set<String> keyset = eveHashMap.keySet();
    
      // 其他校验方法
        // Method method;
        // try {
        //   method = this.getClass().getDeclaredMethod("filter", key, Collector.class);
        //   method.invoke(this, vo, out);
        // } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException
        //     | InvocationTargetException e) {
        //   e.printStackTrace();
        // }
    
  }

  // public void filter(ContractMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(DjkMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(DsfMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(DuebillMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(EtcMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(GrwyMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(GzdfMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(HuanbMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(HuanxMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(@Valid SaMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(SbybMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(SdrqMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(ShopMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

  // public void filter(SjyhMx vo, Collector<Object> out) {
  //   out.collect(vo);
  // }

}
